package mysql

import (
	"database/sql"
	"fmt"
	_ "github.com/go-sql-driver/mysql"
	"miniDW/src/datapack"
	"miniDW/src/interfaces"
	"miniDW/src/log"
	"strings"
	"time"
)

type MysqlSink struct {
	cfg                 *MysqlConfig
	db                  *sql.DB
	sourceName          string
	tableStruct         *datapack.TableStruct
	stmt                *sql.Stmt
	insertSql           string
	insertBatchRowCount int
	count               int
	total               int
	method              string //覆盖模式
	lastTime            time.Time
}

const (
	MaxBufferLength = 8 * 1024 * 1024
)

func (s *MysqlSink) GetBatchInsertSql(num int) string {
	columnNames := s.tableStruct.GetFeildsNames()
	rowScript := "(" + strings.TrimSuffix(strings.Repeat("?,", s.tableStruct.Length()), ",") + ")"
	rowScript = strings.TrimSuffix(strings.Repeat(rowScript+",", num), ",")
	ret := fmt.Sprintf("INSERT INTO %s (%s) VALUES %s ", s.tableStruct.Name(), strings.Join(columnNames, ","), rowScript)
	//fmt.Println(rowScript)
	return ret
}

func (s *MysqlSink) SourceTableStructSetFinished(ctx interfaces.IContext) error {
	var err error
	log.Debug("mysql sink sourceTableStructSetFinished")
	names := ctx.GetSourcesNames()
	if len(names) != 1 {
		return log.Error(fmt.Errorf("MysqlSink not supports more than one Source."))
	}
	sourceTableStruct, err := ctx.GetSourceTableStruct(names[0])
	if err != nil {
		return err
	}
	ts := ctx.GetTableStruct()
	s.tableStruct = sourceTableStruct.Copy()
	s.tableStruct.SetName(ts.Name())
	err = ctx.SetTableStruct(s.tableStruct)
	if err != nil {
		return err
	}

	s.sourceName = names[0]
	err = CreateTableStructure(s.db, s.tableStruct, s.method)
	if err != nil {
		return err
	}

	// 准备插入语句。通过语句批量插入，比通过事务快一个数量级
	s.insertBatchRowCount = MaxBufferLength / (s.tableStruct.ByteLength() + 100)
	if s.insertBatchRowCount > 400 {
		s.insertBatchRowCount = 400
	}
	s.insertSql = s.GetBatchInsertSql(s.insertBatchRowCount)
	s.stmt, err = s.db.Prepare(s.insertSql)
	if err != nil {
		return log.Error(err)
	}

	return nil
}

func (s *MysqlSink) Open(ctx interfaces.IContext) error {
	var err error
	log.Debug("connecting to mysql sink: ", s.cfg.Database)
	s.db, err = Connect(s.cfg)
	if err != nil {
		return err
	}
	s.count = 0
	s.total = 0
	s.lastTime = time.Now()
	log.Debug("connected to mysql sink: ", s.cfg.Database)
	return nil
}

func (s *MysqlSink) Close() error {
	if s.db != nil {
		err := s.db.Close()
		if err != nil {
			return err
		}
	}
	s.db = nil
	if s.stmt != nil {
		_ = s.stmt.Close()
	}
	return nil
}

func (s *MysqlSink) Process(ctx interfaces.IContext) error {
	queue := ctx.GetRowQueue(s.sourceName)
	if queue == nil {
		return log.Error(fmt.Errorf("Query queue %s is nil.", s.sourceName))
	}

	rowCount := queue.Len()
	if rowCount < s.insertBatchRowCount && !queue.IsEOF() {
		return nil
	}

	startTime := time.Now()
	for rowCount > 0 {
		writeRows := min(rowCount, s.insertBatchRowCount)
		rows, err := s.WriteRows(queue, writeRows)
		if err != nil {
			return err
		}
		s.total += rows
		rowCount -= rows
	}
	if queue.IsEOF() {
		ctx.Stop()
	}

	endTime := time.Now()
	spendTime := endTime.Sub(startTime)
	if spendTime > time.Second {
		log.Debug(fmt.Sprintf("total rows: %d, write time: %fs", s.total, spendTime.Seconds()))
	}
	return nil
}

func (s *MysqlSink) WriteRows(queue *datapack.RowQueue, bufCount int) (int, error) {
	var stmt *sql.Stmt
	var err error

	values := make([]interface{}, 0)
	for i := 0; i < bufCount; i++ {
		row := queue.Dequeue()
		//fmt.Printf("row field count: %d\n", len(*row))
		if row == nil {
			return i, log.Error(fmt.Errorf("Query queue %s is nil.", s.sourceName))
		}
		values = append(values, *row...)
	}

	if bufCount < s.insertBatchRowCount {
		insertSql := s.GetBatchInsertSql(bufCount)
		stmt, err = s.db.Prepare(insertSql)
		if err != nil {
			return 0, log.Error(err)
		}
	} else {
		stmt = s.stmt
	}

	_, err = stmt.Exec(values...)
	if err != nil {
		fmt.Printf("values: %d\n", len(values))
		return 0, log.Error(err)
	}

	return bufCount, nil
}

func getMethod(params map[string]interface{}) string {
	method := params["method"].(string)
	return method
}

func CreateMysqlSink(params map[string]interface{}) (interfaces.IComponentFunction, error) {
	var err error
	sink := &MysqlSink{}
	sink.cfg, err = parseConfig(params)
	if err != nil {
		return nil, err
	}
	sink.method = getMethod(params)
	if sink.method == "" {
		sink.method = METHOD_OVERWRITE
	}
	return sink, nil
}
